package com.atguigu.flink.cep;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2022/11/26
 *
 *      套路：
 *              ①有流，基于EventTime，提供水印
 *                      不支持ProcessingTime
 *              ②定义Pattern(模式，匹配规则)
 *              ③调用CEP的方法，把pattern作用到流上，得到一个返回值 模式流
 *              ④从模式流中获取匹配的数据
 */
public class Demo1_HelloWorld
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()   //目前的场景数据都是有序的
                                                      //如何产生水印
                                                      .withTimestampAssigner((e, r) -> e.getTs());

        //①有流，基于EventTime，提供水印
        SingleOutputStreamOperator<WaterSensor> ds = env.readTextFile("data/cep.txt")
                                                       .map(new MapFunction<String, WaterSensor>()
                                                       {
                                                           @Override
                                                           public WaterSensor map(String value) throws Exception {
                                                               String[] words = value.split(",");
                                                               return new WaterSensor(words[0], Long.valueOf(words[1]), Integer.valueOf(words[2]));
                                                           }
                                                       })
                                                       .assignTimestampsAndWatermarks(watermarkStrategy);

        //②定义Pattern
        Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("规则1")
            .where(new SimpleCondition<WaterSensor>()
            {
                //定义匹配规则
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "s1".equals(value.getId());
                }
            });

        //③调用CEP的方法，把pattern作用到流上，得到一个返回值 模式流
        PatternStream<WaterSensor> patternStream = CEP.pattern(ds, pattern);


        // ④从模式流中获取匹配的数据
        patternStream.select(new PatternSelectFunction<WaterSensor, String>()
        {
            /*
                 Map<String, List<WaterSensor>> pattern: 存放的是匹配到达结果
                        key： 定义的规则名    例如:规则1
                        List<WaterSensor>: 匹配到的数据

                      select： 每一条数据的数据，只要负责规则，就调用一次
             */
            @Override
            public String select(Map<String, List<WaterSensor>> map) throws Exception {
                return map.toString();
            }
        })
                     .print().setParallelism(1);


        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }
    }
}
